/***
* ==++==
*
* Copyright (c) Microsoft Corporation.  All rights reserved.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* agnt_tcp_listen.h
*
* C++ Actors: TCP listener
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/

#pragma once

#include <hash_map>
#include <memory>
#include <tuple>

#include "actor_public.h"
#include "comm.h"
#include "context.h"
#include "pid.h"
#include "actor_serialization.h"
#include "uri.h"
#include "xxagents.h"

namespace actors
{
template<typename Element> struct registry_node;

typedef actor *ProcFactory();

struct _process_factory
{
    _process_factory(ProcFactory *function, actor_context context) : Factory(function), Context(context) {}

    ProcFactory *Factory;
    actor_context Context;
};
typedef std::shared_ptr<_process_factory> process_factory;

/// <summary>
/// tcp_listener is one of two classes that handle all the
/// inter-process communication. The dispatcher is listening on 
/// a particular port and accepts incoming messages from a
/// tcp_proxy instance, which is the corresponding outgoing
/// communication class.
///
/// tcp_listener is itself a C++ agent, using the lower-level
/// interface supported in VS 2010. All incoming communication is
/// serialized by this singleton instance.
/// </summary>
class tcp_listener : public Concurrency::agent
{
public:
    /// <summary>
    /// Setup and teardown. Initialization requires a URI identifying a hostname and port.
    /// <remarks>
    /// </summary>
    /// Note initialize and shutdown are not concurrency safe and should only be called once.
    /// </remarks>
    _DISPIMP static tcp_listener * initialize(const std::string &uri = "");
    _DISPIMP static int shutdown();

    /// <summary>
    /// Wait for the dispatcher instance to finish.
    /// </summary>
    _DISPIMP static void wait();

    /// <summary>
    /// Set the compression implementation that is in use across all listeners in the
    /// process.
    /// </summary>
    /// <remarks>
    /// The compression implementation may also do other transformations on
    /// a message body, such as encryption, if it so chooses. It is not limited to
    /// actual compression.
    /// </remarks>
    _DISPIMP static void set_msg_compression(compression *compress) { s_compression = compress; }

    /// <summary>
    /// Default listener instance 
    /// </summary>
    _DISPIMP static tcp_listener *get_default() { return s_default_listener; }

    /// <summary>
    /// The URI that was used to initialize the listener.
    /// </summary>
    _DISPIMP http::uri uri() const { return (this == nullptr) ? http::uri() : m_uri; }

    /// <summary>
    /// Find the local TCP listener that owns the address in question.
    /// </summary>
    _DISPIMP static tcp_listener *find(const http::uri &u);


#pragma region Host and Evict
    /// <summary>
    /// Host a factory for a given actor type. The type name serves in lieu of a relative path.
    /// </summary>
    template<typename ActorType>
    void host()
    {
        const std::string &type_name = typeid(ActorType).name();

        std::stringstream builder;
        builder << "  Hosting factory of '" << type_name.substr(6) << "' at " << m_uri.to_string();
        log::post(LOG_INFO, builder.str());

        process_factory fact(new _process_factory(ActorType::factory, actor_context()));
        _register_factory(type_name, fact);
        actor::add_to_directory<ActorType>(m_uri);
    }

    template<typename ActorType>
    void host(actor_context context)
    {
        const std::string &type_name = typeid(ActorType).name();

        std::stringstream builder;
        builder << "  Hosting factory of '" << type_name.substr(6) << "' at " << m_uri;
        log::post(LOG_INFO, builder.str());

        process_factory fact(new _process_factory(ActorType::factory, context));
        _register_factory(type_name, fact);
        actor::add_to_directory<ActorType>(m_uri);
    }

    /// <summary>
    /// Host a factory or instance of a given type at a given relative address.
    /// </summary>
    template<typename ActorType>
    void host(const std::string &resource)
    {
        const std::string &type_name = typeid(ActorType).name();

        http::uri uri = m_uri.append_path(resource);
        std::stringstream builder;
        builder << "  Hosting factory of '" << type_name.substr(6) << "' at " << uri.to_string();
        log::post(LOG_INFO, builder.str());

        process_factory fact(new _process_factory(ActorType::factory, actor_context()));
        _register_factory(type_name, fact);
        _register_factory(resource, fact);
        actor::add_to_directory<ActorType>(uri);
    }

    template<typename ActorType>
    void host(const std::string &resource, actor_context context)
    {
        const std::string &type_name = typeid(ActorType).name();

        http::uri uri = m_uri.append_path(resource);
        std::stringstream builder;
        builder << "  Hosting factory of '" << type_name.substr(6) << "' at " << uri.to_string();
        log::post(LOG_INFO, builder.str());

        process_factory fact(new _process_factory(ActorType::factory, context));
        _register_factory(type_name, fact);
        _register_factory(resource, fact);
        actor::add_to_directory<ActorType>(uri);
    }

    // TFS: # 336255
    // TODO: when host_instance<T>() is used, multiple PIDS are all mapped to the
    //       same single instance. When the server is restarted, they don't find the
    //       new instance unless they send a new CMD_SPAWN -- fix this.

    /// <summary>
    /// Host an actor instance for a given actor type. The type name
    /// serves in lieu of a relative path.
    /// </summary>
    template<typename ActorType>
    PID host_instance()
    {
        const std::string &type_name = typeid(ActorType).name();

        std::stringstream builder;
        builder << "  Hosting instance of '" << type_name.substr(6) << "' at " << m_uri;
        log::post(LOG_INFO, builder.str());

        actor *process = ActorType::factory();
        if ( process == nullptr ) return PID();

        _register_instance(type_name, process);
        process->set_context(this, actor_context());
        process->initialize(actor::ACTR_Create);
        process->start();

        return process->self();
    }

    template<typename ActorType>
    PID host_instance(actor_context context)
    {
        const std::string &type_name = typeid(ActorType).name();

        std::stringstream builder;
        builder << "  Hosting instance of '" << type_name.substr(6) << "' at " << m_uri;
        log::post(LOG_INFO, builder.str());

        actor *process = ActorType::factory();
        if ( process == nullptr ) return PID();

        _register_instance(type_name, process);
        process->set_context(this,context);
        process->initialize(actor::ACTR_Create);
        process->start();

        return process->self();
    }

    /// <summary>
    /// Host an actor instance for a given actor type at a given relative address.
    /// </summary>
    template<typename ActorType>
    PID host_instance(const std::string &resource)
    {
        const std::string &type_name = typeid(ActorType).name();

        http::uri uri = m_uri.append_path(resource);
        std::stringstream builder;
        builder << "  Hosting instance of '" << type_name.substr(6) << "' at " << uri.to_string();
        log::post(LOG_INFO, builder.str());

        actor *process = ActorType::factory();
        if ( process == nullptr ) return PID();

        _register_instance(resource, process);
        process->set_context(this, actor_context());
        process->initialize(actor::ACTR_Create);
        process->start();
        return process->self();
    }

    template<typename ActorType>
    PID host_instance(const std::string &resource, actor_context context)
    {
        const std::string &type_name = typeid(ActorType).name();

        std::stringstream builder;
        builder << "  Hosting instance of '" << type_name.substr(6) << "' at " << (m_uri+resource);
        log::post(LOG_INFO, builder.str());

        actor *process = ActorType::factory();
        if ( process == nullptr ) return PID();

        _register_instance(resource, process);
        process->set_context(this,context);
        process->initialize(actor::ACTR_Create);
        process->start();
        return process->self();
    }

    /// <summary>
    /// Remove any factories and instances hosted by the listener.
    /// </summary>
    template<typename ActorType>
    void evict()
    {
        std::string type_name = typeid(ActorType).name();

        std::stringstream builder;
        builder << "  Evicting factories and instances of '" << type_name.substr(6) << "' at " << m_uri.to_string();
        log::post(LOG_INFO, builder.str());

        _unregister_factory(type_name);
        _unregister_instance(type_name);
        actor::remove_from_directory<ActorType>(m_uri);
    }

    /// <summary>
    /// Remove any factories and instances hosted by the listener at the given relative address.
    /// </summary>
    template<typename ActorType>
    void evict(const std::string &resource)
    {
        std::string type_name = typeid(ActorType).name();

        http::uri uri = m_uri.append_path(resource);
        std::stringstream builder;
        builder << "  Evicting factories and instances of '" << type_name.substr(6) << "' at " << uri.to_string();
        log::post(LOG_INFO, builder.str());
        _unregister_factory(type_name);
        _unregister_factory(resource);
        _unregister_instance(resource);
        actor::remove_from_directory<ActorType>(m_uri);
        actor::remove_from_directory<ActorType>(uri);
    }
#pragma endregion

protected:
    void run();

#pragma region "private declarations"
private:
    /// <summary>
    ///
    /// </summary>
    tcp_listener(const http::uri &uri);
    ~tcp_listener();

    friend class tcp_proxy;
    friend class PID;
    friend class actor;
    friend class _actor_info;

    _DISPIMP friend bool ::deserialize(PID &pid, actors::Serializer &serializer);

    _DISPIMP process_factory _get_factory(const std::string &uri);
    _DISPIMP void _register_factory(const std::string &uri, process_factory );
    _DISPIMP void _unregister_factory(const std::string &uri);

    _DISPIMP actor *_get_instance(const std::string &uri);
    _DISPIMP void _register_instance(const std::string &uri, actor *);
    _DISPIMP void _unregister_instance(const std::string &uri);

    _DISPIMP static actor_info _get_actor(const msg_guid &id);
    _DISPIMP static void _register_actor(const msg_guid &id, bool once_only, actor_info process);
    _DISPIMP static void _unregister_actor(actor_info );

    bool _setup();
    void _handle_conn(SOCKET ClientSocket);
    msg_op _finish_msg(SOCKET ClientSocket, dispatch_msg *msg);

    SOCKET m_socket;

    static tcp_listener *s_default_listener;

    http::uri m_uri;

    Concurrency::single_assignment<bool> m_startup_status;

    static std::vector<tcp_listener *> s_listeners;

    Concurrency::critical_section m_local_lock;

    static compression *s_compression;

    static Concurrency::critical_section s_global_lock;

    static std::hash_map<msg_guid, std::tuple<bool, actor_info>> s_processes;

    std::vector<actor_info> process_cache;

    registry_node<process_factory> *factory_registry;
    registry_node<actor *> *instance_registry;
#pragma endregion
};

} // namespace actors